package com.landleaf.smarthome.util.rxutils;

/**
 * Created by 海航
 * 2019/8/14
 * 功能描述:RxJava工具
 */


import com.landleaf.smarthome.util.LogUtils;
import com.trello.rxlifecycle3.android.ActivityEvent;
import com.trello.rxlifecycle3.android.FragmentEvent;
import com.trello.rxlifecycle3.components.support.RxAppCompatActivity;
import com.trello.rxlifecycle3.components.support.RxFragment;

import java.util.concurrent.TimeUnit;

import androidx.annotation.NonNull;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.FlowableTransformer;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.observers.DisposableObserver;

import static io.reactivex.BackpressureStrategy.LATEST;


public final class RxJavaUtils {

    private final static String TAG = "RxJavaUtils";

    private RxJavaUtils() {
        throw new UnsupportedOperationException("u can't instantiate me...");
    }

    /**
     * 生成Observable
     *
     * @param <T>
     * @return
     */
    public static <T> Observable<T> createObservable(final T t) {
        return Observable.create(new ObservableOnSubscribe<T>() {
            @Override
            public void subscribe(ObservableEmitter<T> emitter) throws Exception {
                try {
                    emitter.onNext(t);
                    emitter.onComplete();
                } catch (Throwable e) {
                    emitter.onError(e);
                }
            }
        });
    }

    public static <T> ObservableTransformer<T, T> composeFragment(final RxFragment fragment) {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> upstream) {
                return upstream.compose(RxSchedulerUtils._io_main_o())
                        .compose(fragment.bindUntilEvent(FragmentEvent.DESTROY));

            }
        };
    }

    public static <T> ObservableTransformer<T, T> composeAct(final RxAppCompatActivity activity) {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> upstream) {
                return upstream.compose(RxSchedulerUtils._io_main_o())
                        .compose(activity.bindUntilEvent(ActivityEvent.DESTROY));

            }
        };
    }

    /**
     * 生成Observable
     *
     * @param <T>
     * @return
     */
    public static <T> Flowable<T> createFlowable(final T t) {
        return Flowable.create(new FlowableOnSubscribe<T>() {
            @Override
            public void subscribe(FlowableEmitter<T> emitter) throws Exception {
                try {
                    emitter.onNext(t);
                    emitter.onComplete();
                } catch (Throwable e) {
                    emitter.onError(e);
                }
            }
        }, LATEST);

    }

    //========================轮询操作==========================//

    /**
     * 轮询操作
     *
     * @param interval 轮询间期
     */
    public static Flowable<Long> polling(long interval) {
        return polling(0, interval, TimeUnit.SECONDS);
    }
    public static Flowable<Long> polling(long interval, @NonNull TimeUnit unit) {
        return polling(0, interval, unit);
    }
    /**
     * 轮询操作
     *
     * @param interval 轮询间期
     * @param consumer 监听事件
     */
    public static Disposable polling(long interval, @NonNull Consumer<Long> consumer) {
        return polling(0, interval, consumer);
    }

    /**
     * 轮询操作
     *
     * @param initialDelay 初始延迟
     * @param interval     轮询间期
     * @param consumer     监听事件
     */
    public static Disposable polling(long initialDelay, long interval, @NonNull Consumer<Long> consumer) {
        return polling(initialDelay, interval, TimeUnit.SECONDS, consumer, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                LogUtils.d("轮询操作异常:" + throwable.getMessage());
            }
        });
    }

    /**
     * 轮询操作
     *
     * @param initialDelay 初始延迟
     * @param interval     轮询间期
     * @param unit         轮询间期时间单位
     */
    public static Flowable<Long> polling(long initialDelay, long interval, @NonNull TimeUnit unit) {
        return Flowable.interval(initialDelay, interval, unit)
                .observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * 轮询操作
     *
     * @param initialDelay  初始延迟
     * @param interval      轮询间期
     * @param unit          轮询间期时间单位
     * @param consumer      监听事件
     * @param errorConsumer 出错的事件
     */
    public static Disposable polling(long initialDelay, long interval, @NonNull TimeUnit unit, @NonNull Consumer<Long> consumer, @NonNull Consumer<Throwable> errorConsumer) {
        return Flowable.interval(initialDelay, interval, unit)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer, errorConsumer);
    }

    //=================倒计时=================//

    /**
     * 倒计时操作【间隔1秒】
     *
     * @param totalTime  倒计时总时长
     * @param subscriber 事件订阅
     * @return
     */
    public static Disposable countDown(final long totalTime, @NonNull DisposableObserver<Long> subscriber) {
        return countDown(totalTime, 1, TimeUnit.SECONDS)
                .subscribeWith(subscriber);
    }

    /**
     * 倒计时操作【间隔1秒】
     *
     * @param totalTime 倒计时总时长
     * @return
     */
    public static Observable<Long> countDown(final long totalTime) {
        return countDown(totalTime, 1, TimeUnit.SECONDS);
    }

    /**
     * 倒计时操作
     *
     * @param totalTime 倒计时总时长
     * @param interval  倒计时间隔
     * @param unit      时间间隔单位
     * @return
     */
    public static Observable<Long> countDown(final long totalTime, long interval, @NonNull TimeUnit unit) {
        return Observable.interval(0, interval, unit)
                .take((int) Math.floor(totalTime / (double) interval) + 1)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long time) throws Exception {
                        return totalTime - time;
                    }
                })
                .compose(RxSchedulerUtils.<Long>_io_main_o());
    }

    public static Observable<Long> interval(final long totalTime, long interval, @NonNull TimeUnit unit) {
        return Observable.interval(0, interval, unit)
                .take(totalTime)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long time) throws Exception {
                        return time;
                    }
                })
                .compose(RxSchedulerUtils.<Long>_io_main_o());
    }
    //========================延迟操作==========================//


    /**
     * 延迟操作
     *
     * @param delayTime     延迟时间
     * @param unit          延迟时间单位
     * @param consumer      监听事件
     * @param errorConsumer 出错的事件
     */
    public static Disposable delay(@NonNull long delayTime, @NonNull TimeUnit unit, @NonNull Consumer<Long> consumer, @NonNull Consumer<Throwable> errorConsumer) {
        return Flowable.timer(delayTime, unit)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer, errorConsumer);
    }

    /**
     * 延迟操作
     *
     * @param delayTime 延迟时间
     * @param unit      延迟时间单位
     */
    public static Observable<Long> delay(long delayTime, @NonNull TimeUnit unit) {
        return Observable.timer(delayTime, unit)
                .observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * @param delayTime
     * @param unit
     * @return
     */
    public static Observable<Long> delay(long delayTime, @NonNull TimeUnit unit, Scheduler scheduler) {
        return Observable.timer(delayTime, unit, scheduler)
                .observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * 延迟操作
     *
     * @param delayTime  延迟时间
     * @param unit       延迟时间单位
     * @param subscriber 订阅的事件
     */
    public static Disposable delay(long delayTime, @NonNull TimeUnit unit, @NonNull DisposableObserver<Long> subscriber) {
        return Observable.timer(delayTime, unit)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(subscriber);
    }

    /**
     * 延迟操作
     *
     * @param t         发射源
     * @param delayTime 延迟时间
     * @param unit      延迟时间单位
     */
    public static <T> Observable<T> delay(@NonNull T t, long delayTime, @NonNull TimeUnit unit) {
        return Observable.just(t)
                .delay(delayTime, unit)
                .observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * 延迟操作
     *
     * @param t          发射源
     * @param delayTime  延迟时间
     * @param unit       延迟时间单位
     * @param subscriber 订阅的事件
     */
    public static <T> Disposable delay(@NonNull T t, long delayTime, @NonNull TimeUnit unit, @NonNull DisposableObserver<T> subscriber) {
        return Observable.just(t)
                .delay(delayTime, unit)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(subscriber);
    }


    //==================Transformer=====================//

    /**
     * 执行异步任务（IO线程处理，UI线程显示）
     *
     * @param t           处理入参
     * @param transformer 转化器
     * @return
     */
    public static <T, R> Flowable<R> executeAsyncTask(@NonNull T t, @NonNull FlowableTransformer<T, R> transformer) {
        return Flowable.just(t)
                .compose(transformer)
                .compose(RxSchedulerUtils.<R>_io_main_f());
    }

    /**
     * 执行异步任务（IO线程处理，UI线程显示）
     *
     * @param t           处理入参
     * @param transformer 转化器
     * @return
     */
    public static <T, R> Observable<R> executeAsyncTask(@NonNull T t, @NonNull ObservableTransformer<T, R> transformer) {
        return Observable.just(t)
                .compose(transformer)
                .compose(RxSchedulerUtils.<R>_io_main_o());
    }


    /**
     * 执行异步任务（IO线程处理，UI线程显示）
     *
     * @param t             处理入参
     * @param transformer   转化器
     * @param consumer      监听事件
     * @param errorConsumer 出错的事件
     * @return
     */
    public static <T, R> Disposable executeAsyncTask(@NonNull T t, @NonNull FlowableTransformer<T, R> transformer, @NonNull Consumer<R> consumer, @NonNull Consumer<Throwable> errorConsumer) {
        return executeAsyncTask(t, transformer).subscribe(consumer, errorConsumer);
    }

    /**
     * 执行异步任务（IO线程处理，UI线程显示）
     *
     * @param t           处理入参
     * @param transformer 转化器
     * @param subscriber  订阅事件
     * @return
     */
    public static <T, R> Disposable executeAsyncTask(@NonNull T t, @NonNull ObservableTransformer<T, R> transformer, @NonNull DisposableObserver<R> subscriber) {
        return executeAsyncTask(t, transformer).subscribeWith(subscriber);
    }

    //----------------绑定生命周期--------------------------------
    public static <T> ObservableTransformer<T, T> comActCompose(final RxAppCompatActivity activity) {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> observable) {
                return observable
                        .compose(RxSchedulerUtils._io_main_o())
                        .compose(activity.<T>bindUntilEvent(ActivityEvent.DESTROY))
                        ;
            }
        };
    }

    public static <T> ObservableTransformer<T, T> comFrgCompose(final RxFragment fragment) {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> observable) {
                return observable
                        .compose(RxSchedulerUtils._io_main_o())
                        .compose(fragment.<T>bindUntilEvent(FragmentEvent.DESTROY))
                        ;
            }
        };
    }


}